package mammuthus.master

import com.google.inject.{Singleton, Inject}
import net.csdn.common.settings.Settings
import akka.actor.{ActorRef, Props}
import scala.concurrent._
import akka.pattern.ask
import scala.concurrent.duration._
import akka.util.Timeout
import mammuthus.slave.{InteractiveShellCommand, ShellCommand}
import net.csdn.common.logging.Loggers
import scala.collection.mutable

import java.util.concurrent.TimeUnit
import mammuthus.master.app.AppOperation
import mammuthus.master.docker.DockerOperation
import mammuthus.tool.SystemTool
import mammuthus.protocal._
import mammuthus.docker.protocal.TimeOutResponse
import mammuthus.protocal.WriteFileTaskResponse
import mammuthus.protocal.ExecuteDownloadCommandRequest
import mammuthus.protocal.ExecuteMammuthusConfStartRequest
import scala.Some
import mammuthus.protocal.ExecuteBlockShellCommandResponse
import mammuthus.protocal.WriteFileTask
import mammuthus.protocal.BlockExecuteDownloadCommandRequest
import mammuthus.protocal.ShellTask
import mammuthus.protocal.ExecuteDownloadCommandResponse
import mammuthus.protocal.ExecuteBlockShellCommandRequest
import mammuthus.protocal.ExecuteShellCommandRequest
import mammuthus.protocal.ShellProgressTask
import mammuthus.protocal.ExecuteShellCommandResponse
import mammuthus.protocal.ExecuteShellCommandProgressResponse
import mammuthus.protocal.AgentInfo


/**
 * 3/24/15 WilliamZhu(allwefantasy@gmail.com)
 */
@Singleton
class Master @Inject()(conf: Settings, akkaUtil: Akkautil2) extends AppOperation with DockerOperation with TokenGenerate {

  val logger = Loggers.getLogger(getClass)
  val name = "Master"
  implicit val timeout = Timeout(60 seconds)
  val masterHost = sys.env.get("MAMMUTHUS_MASTER_HOST") match {
    case Some(i) => i
    case None => conf.get("master.host", SystemTool.hostName match {
      case Some(i) => i
      case None => ShellCommand.exec("hostname")
    })
  }

  val masterPort = sys.env.get("MAMMUTHUS_MASTER_PORT") match {
    case Some(i) => i.toInt
    case None => conf.get("master.port", "5010").toInt
  }


  val systemName = conf.get("eq.actor-system.name", "mammuthus")

  val (system, akkaPort) = akkaUtil.createActorSystem(masterHost, masterPort)

  val masterUrl = s"akka.tcp://${systemName}@${masterHost}:${masterPort}"

  val masterActor = system.actorOf(Props(classOf[MasterActor], this), name)

  val slaves = new mutable.HashMap[String, (ActorRef, AgentInfo)]() with mutable.SynchronizedMap[String, (ActorRef, AgentInfo)]
  val offLineSlaves = new mutable.HashMap[String, (ActorRef, AgentInfo)]() with mutable.SynchronizedMap[String, (ActorRef, AgentInfo)]
  val scheduler = system.scheduler

  val masterHttpPort = _conf.get("http.port", "9003")

  def cleanTimeRange = _conf.getAsInt("mammuthus.master.slave_expire_interval",30)

  implicit val executor = system.dispatcher

  val task = new Runnable {
    def run() {
      logger.info("master task check slave alive or dead")
      slaves.filter {
        f =>
          (System.currentTimeMillis() - f._2._2.heartBeatTime) / (1000 ) > cleanTimeRange
      }.foreach {
        f =>
          logger.info(s"agent ${f._1} is offline")
          offLineSlaves.put(f._1, slaves.remove(f._1).get)
      }
    }
  }
  scheduler.schedule(
    initialDelay = Duration(5, TimeUnit.SECONDS),
    interval = Duration(5, TimeUnit.SECONDS),
    runnable = task)

  def _conf = conf

  def _remoteExecute[T](hostNames: String, f: (ActorRef, String) => Awaitable[Any]): scala.collection.mutable.ArraySeq[T] = {

    hostNames.split(",").filter(f=>slaves.contains(f)).map {
      hostName =>
        val ar = slaves(hostName)._1
        f(ar, hostName)
    }.map {
      f =>
        Await.result(f, timeout.duration)
    }.filterNot {
      f =>
        val re = f.isInstanceOf[TimeOutResponse]
        if (re) logger.info("有服务器响应超时")
        re
    }.map {
      ecr =>
        ecr.asInstanceOf[T]
    }
  }

  def executeShellCommand(hostNames: String, shellCommand: String, tt: Long) = {

    val token = tokenGenerate
    _remoteExecute[ExecuteShellCommandResponse](hostNames, (ar, hostName) => {
      ar ? ExecuteShellCommandRequest(hostName, shellCommand, token, tt)
    })
  }

  def executeBlockShellCommand(hostNames: String, shellCommand: String, tt: Long = InteractiveShellCommand.defaultTimeOut) = {

    val token = tokenGenerate

    _remoteExecute[ExecuteBlockShellCommandResponse](hostNames, (ar, hostName) => {
      ar ? ExecuteBlockShellCommandRequest(ExecuteShellCommandRequest(hostName, shellCommand, token, tt))
    })

  }

  def startMammuthusConf(hostNames: String) = {
    val servers = appInstallHosts("zookeeper").split(",").map(f=>findIpByHost(f)).filter(f=>f != "-").mkString(",")
    _remoteExecute[ExecuteMammuthusConfStartResponse](hostNames, (ar, hostName) => {
      ar ? ExecuteMammuthusConfStartRequest(servers)
    })
  }

  def writeFile(hostNames: String, msg: String, path: String, append: Boolean = true) = {

    logger.info(s"writeFile $path")

    _remoteExecute[WriteFileTaskResponse](hostNames, (ar, hostName) => {
      ar ? WriteFileTask(hostName, msg, path, append)
    }).toList

  }

  def writeJSONFile(hostNames: String, msg:Map[String,String], path: String, append: Boolean = true) = {

    logger.info(s"writeFile $path")

    _remoteExecute[WriteFileTaskResponse](hostNames, (ar, hostName) => {
      ar ? WriteJSONFileTask(hostName, msg, path)
    }).toList

  }

  def download(hostNames: String, url: String, filePath: String) = {

    hostNames.split(",").map {
      hostName =>
        val ar = slaves(hostName)._1
        ar ! ExecuteDownloadCommandRequest(hostName, url, filePath)
    }
  }

  def blockDownload(hostNames: String, url: String, filePath: String) = {

    _remoteExecute[ExecuteDownloadCommandResponse](hostNames, (ar, hostName) => {
      ar ? BlockExecuteDownloadCommandRequest(hostName, url, filePath)
    })

  }

  def shellCommandProcess(hostNames: String, taskId: String, offset: Long) = {
    _remoteExecute[ExecuteShellCommandProgressResponse](hostNames, (ar, hostName) => {
      ar ? ShellProgressTask(ShellTask(taskId, hostName), offset)
    })
  }

  def getCCParams(cc: AnyRef) =
    (Map[String, Any]() /: cc.getClass.getDeclaredFields) {(a, f) =>
      f.setAccessible(true)
      a + (f.getName -> f.get(cc))
  }
}
